package mc.support.zmq.mla;

import mc.support.proto.*;
import mc.support.util.PlainUtil;
import mc.support.zmq.ProtoConstants;
import mc.support.zmq.err.ProtoException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created with IntelliJ IDEA. <br/>
 * Author: Francis Yuen    <br/>
 * Date: 2014-10-28  <br/>
 */
public abstract class MLAWorker implements Runnable, MLAConstants, ProtoConstants {

    private static final Logger log = LoggerFactory.getLogger( MLAWorker.class );

    private ZMQ.Context context;
    private BizHandler<GeneralReq, GeneralRsp> generalHandler;
    private AtomicBoolean keepOn;

    public MLAWorker setGeneralHandler( BizHandler<GeneralReq, GeneralRsp> generalHandler ) {
        this.generalHandler = generalHandler;
        return this;
    }

    public MLAWorker setContext( ZMQ.Context context ) {
        this.context = context;
        return this;
    }

    public MLAWorker setKeepOn( AtomicBoolean keepOn ) {
        this.keepOn = keepOn;
        return this;
    }

    abstract void prepared();

    @Override
    public void run() {
        final ZMQ.Socket socket = context.socket( ZMQ.REP );
        socket.connect( "inproc://" + MLA_WORKERS );
        prepared();

        byte[] inBs, outBs;
        Biz biz = null;
        while ( keepOn.get() ) {
            try {
                inBs = socket.recv( 0 );
                GeneralReq req = GeneralReq.parseFrom( inBs );
                biz = req.getBiz();
                outBs = generalHandler.handle( req ).toByteArray();
                socket.send( outBs, 0 );
            } catch ( Exception err ) {
                socket.send( errorRsp( err, biz ) );
                log.info( "error when handling request: {} ", PlainUtil.getStackTrace( err ) );
            }
        }
    }

    private byte[] errorRsp( Exception err, Biz biz ) {
        String msg = err.getMessage();
        int errorNo;
        if ( err instanceof ProtoException ) {
            errorNo = ( (ProtoException) err ).getErrorNo();
        } else {
            errorNo = ProtoException.UNEXPECTED_ERROR;
        }
        return GeneralRsp.newBuilder()
                .setMagicCode( MAGIC_CODE )
                .setErrorNo( errorNo )
                .setBiz( null != biz ? biz : Biz.SYS_STUB )
                .setMsg( null != msg ? msg : "" )
                .build().toByteArray();
    }

}
